package com.bigdata.kafkastream;

import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.processor.TopologyBuilder;

import java.util.Properties;

public class Application {
    public static void main(String[] args) {
        //从log Topic  ==> recommender
        //1.基础设置
        String brokers = "bigdata211:9092";
        String zookeepers = "bigdata211:2181";

        String from = "log";
        String to = "recommender";

        //4.创建Properties
        Properties props = new Properties();
        //服务名
        props.put(StreamsConfig.APPLICATION_ID_CONFIG,"StreamConfig");
        //kafka集群地址
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,brokers);
        //zookeeper集群地址
        props.put(StreamsConfig.ZOOKEEPER_CONNECT_CONFIG,zookeepers);
        //时间戳问题（从flume传输数据没有是时间戳）【保留】
        props.put(StreamsConfig.TIMESTAMP_EXTRACTOR_CLASS_CONFIG,MyEvenTime.class);


        //3.创建StreamConfig//流配置
        StreamsConfig config = new StreamsConfig(props);

        //2.创建kafkaStram的拓部结构
        TopologyBuilder builder = new TopologyBuilder();
           //源文件
        builder.addSource("SOURCE",from)
                 //上一层
                .addProcessor("PROCESSOR", ()->new LogProcessor(), "SOURCE")//过滤层
                 //目的文件
                .addSink("SINK",to,"PROCESSOR");

        //反推参数：builder，config
        KafkaStreams streams = new KafkaStreams(builder,config);

        streams.start();
        System.out.println("KafkaStream GO !");
    }
}
